package drds.binlog.parse.index;

import drds.binlog.common.position.LogPosition;
import drds.binlog.common.utils.JsonUtils;
import drds.binlog.common.zookeeper.ZooKeeperClient;
import drds.binlog.common.zookeeper.Zookeepers;
import drds.binlog.parse.exception.BinlogParseException;
import org.I0Itec.zkclient.exception.ZkNoNodeException;


public class ZooKeeperLogPositionManager extends AbstractLogPositionManager
{

    private final ZooKeeperClient zooKeeperClient;

    public ZooKeeperLogPositionManager(ZooKeeperClient zooKeeperClient)
    {
        if (zooKeeperClient == null)
        {
            throw new NullPointerException("null zkClient");
        }

        this.zooKeeperClient = zooKeeperClient;
    }

    @Override
    public LogPosition getLatestIndexBy(String destination)
    {
        String path = Zookeepers.getParsePath(destination);
        byte[] data = zooKeeperClient.readData(path, true);
        if (data == null || data.length == 0)
        {
            return null;
        }

        return JsonUtils.unmarshalFromByte(data, LogPosition.class);
    }

    @Override
    public void persistLogPosition(String destination, LogPosition logPosition) throws BinlogParseException
    {
        String path = Zookeepers.getParsePath(destination);
        byte[] data = JsonUtils.marshalToByte(logPosition);
        try
        {
            zooKeeperClient.writeData(path, data);
        } catch (ZkNoNodeException e)
        {
            zooKeeperClient.createPersistent(path, data, true);
        }
    }

}
